package com.swak.reactivex.transport;

import java.util.Collection;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CopyOnWriteArraySet;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.exception.RemotingException;

/**
 * ExchangeHandlerDispatcher.
 * 
 * @author 618lf
 */
public class ExchangeHandlerDispatcher implements ExchangeHandler {

	private static final Logger logger = LoggerFactory.getLogger(ExchangeHandlerDispatcher.class);

	private final Collection<ExchangeHandler> channelHandlers = new CopyOnWriteArraySet<ExchangeHandler>();

	public ExchangeHandlerDispatcher addChannelHandler(ExchangeHandler handler) {
		this.channelHandlers.add(handler);
		return this;
	}

	public ExchangeHandlerDispatcher removeChannelHandler(ExchangeHandler handler) {
		this.channelHandlers.remove(handler);
		return this;
	}

	@Override
	public void connected(Channel channel) throws RemotingException {
		for (ChannelHandler listener : channelHandlers) {
			try {
				listener.connected(channel);
			} catch (Throwable t) {
				logger.error(t.getMessage(), t);
			}
		}
	}

	@Override
	public void disconnected(Channel channel) throws RemotingException {
		for (ChannelHandler listener : channelHandlers) {
			try {
				listener.disconnected(channel);
			} catch (Throwable t) {
				logger.error(t.getMessage(), t);
			}
		}
	}

	@Override
	public void sent(Channel channel, Object message) throws RemotingException {
		for (ChannelHandler listener : channelHandlers) {
			try {
				listener.sent(channel, message);
			} catch (Throwable t) {
				logger.error(t.getMessage(), t);
			}
		}
	}

	@Override
	public void received(Channel channel, Object message) throws RemotingException {
		for (ChannelHandler listener : channelHandlers) {
			try {
				listener.received(channel, message);
			} catch (Throwable t) {
				logger.error(t.getMessage(), t);
			}
		}
	}

	@Override
	public void caught(Channel channel, Throwable exception) throws RemotingException {
		for (ChannelHandler listener : channelHandlers) {
			try {
				listener.caught(channel, exception);
			} catch (Throwable t) {
				logger.error(t.getMessage(), t);
			}
		}
	}

	@Override
	public CompletionStage<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
		return this.channelHandlers.stream().findFirst().get().reply(channel, message);
	}
}
